package org.sg20.labelpropagation;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.util.LineReader;
import org.sg20.labelpropagation.lpainiter.LpaIniterDriver;
import org.sg20.labelpropagation.lpaiter.LpaIterDriver;
import org.sg20.labelpropagation.lpaviewer.LpaViewerDriver;

import java.io.IOException;

public class LpaDriver {

    private FileSystem fs;
    // 将一个文件的内容拷贝到另一个文件中
    public static void copyFile(String fsName,String from_path, String to_path) throws IOException {
        Path path_from = new Path(from_path);
        Path path_to = new Path(to_path);
        // 获取HDFS 文件系统
        Configuration conf = new Configuration();
//        conf.set("fs.defaultFS",fsName);
        FileSystem fs=FileSystem.get(conf);
        FSDataInputStream inputStream = fs.open(path_from);
        LineReader lineReader = new LineReader(inputStream, conf);
        FSDataOutputStream outputStream = fs.create(path_to);
        Text line = new Text();
        while (lineReader.readLine(line) > 0) {
            String str = line + "\n";
            outputStream.write(str.getBytes());
        }
        lineReader.close();
        outputStream.close();
    }
    /**
     *
     * @param args 文件系统(HDFS or Local),链接文件(图文件),标签数目,初始标签文件,输出目录,输出目录
     */
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        String fsName=args[0];
        String graphFile=args[1];
        int labelNum=Integer.parseInt(args[2]);
        int propagateNum=Integer.parseInt(args[3]);
        String initDotsFile=args[4];
        String outPath=args[5];

        int epoch=0;
        String curPath=outPath+"/epoch"+epoch;
        LpaIniterDriver.run(fsName,graphFile,labelNum,initDotsFile,curPath+"/iter");
        String curMD5= LpaViewerDriver.run(epoch,fsName,curPath+"/iter",propagateNum,curPath+"/viewer");
        String preMD5="";
        String prePath;
        while (!preMD5.equals(curMD5) && epoch <= 10) {
            prePath = curPath;
            epoch += 1;
            curPath = outPath + "/epoch" + epoch;
            preMD5 = curMD5;
            LpaIterDriver.run(epoch, fsName, prePath + "/iter", labelNum, propagateNum, initDotsFile, curPath + "/iter");
            curMD5 = LpaViewerDriver.run(epoch, fsName, curPath + "/iter", propagateNum, curPath + "/viewer");
        }
        //最后一轮讲标签数量置1,并拷贝最终结果
        epoch+=1;
        prePath=curPath;
        curPath=outPath+"/epoch"+epoch;
        LpaViewerDriver.run(epoch,fsName,prePath+"/iter",1,curPath+"/viewer");
        copyFile(fsName,curPath+"/viewer/part-r-00000",outPath+"/lpa_res.txt");
    }
}
